package com.lxn.until

import org.apache.kafka.common.TopicPartition

import java.util
import org.apache.spark.streaming.kafka010.OffsetRange

import scala.collection.mutable

/**
 * Offset管理工具类， 用于往redis中存储和读取offset
 *
 * 管理方案:
 *   1.  后置提交偏移量  ->  手动控制偏移量提交
 *   2.  手动控制偏移量提交 ->  SparkStreaming提供了手动提交方案，但是我们不能用，因为我们会对DStream的结构进行转换.
 *   3.  手动的提取偏移量维护到redis中
 *         -> 从kafka中消费到数据,先提取偏移量
 *         -> 等数据成功写出后，将偏移量存储到redis中
 *         -> 从kafka中消费数据之前，先到redis中读取偏移量， 使用读取到的偏移量到kakfa中消费数据
 *
 *   4. 手动的将偏移量存储到redis中，每次消费数据需要使用存储的offset进行消费，每次消费数据后，要将本次消费的offset存储到redis中。
 */

object MyOffsetUtils {

  /**
   * 往Redis中存储offset
   * 问题： 存的offset从哪来？
   *            从消费到的数据中提取出来的，传入到该方法中。
   *            offsetRanges: Array[OffsetRange]
   *        offset的结构是什么？
   *            Kafka中offset维护的结构
   *               groupId + topic + partition => offset
   *            从传入进来的offset中提取关键信息
   *        在redis中怎么存?
   *          类型: hash
   *          key : groupId + topic
   *          value: partition - offset  ， partition - offset 。。。。
   *          写入API: hset / hmset
   *          读取API: hgetall
   *          是否过期: 不过期
   */

  //saveOffset
  def saveOffset(groupId : String, topic : String,offsetRanges:Array[OffsetRange]): Unit = {
    //我们存储offset
    //在kafka里面,它是按照组加主题，分区，偏移量进行存储，这里我们使用hash类型存储，一次性将数据存储
    if(offsetRanges!=null && offsetRanges.length>0){
      //使用这个对获取到的分区和偏移量进行封装
      val offset : util.HashMap[String,String] = new util.HashMap[String, String]()
      for(offsetRange <- offsetRanges){
        //开始获取值
        val partition = offsetRange.partition
        val endOffset = offsetRange.untilOffset
        offset.put(partition.toString,endOffset.toString)
      }
      //定义key并按照redis的存储格式 key - value 格式进行存储
      val redKey = s"offsets:$groupId:$topic";
      //通过工具类获取redis连接对象，然后进行批量存储
      val jedis = MyRedisUtils.getRedisFromPool()
      jedis.hmset(redKey,offset)
      //关闭连接对象
      jedis.close()
    }
  }

  //写完了存储，我们写读取，在redis中，读取数据是根据key读取，所以我们需要传入key
  def readOffset(groupId : String, topic : String): Map[TopicPartition,Long] = {
    //现在我们想要获取数据，所以得先连接
    val jedis = MyRedisUtils.getRedisFromPool()
    //拼接key
    val redisKey = s"offsets:$groupId:$topic"
    val offsets :util.Map[String,String] = jedis.hgetAll(redisKey)
    println("222读取到offset: "+offsets)
    //我们封装最后的读取结果
    val result: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition, Long]()
    //现在，我们需要将数据封装 但是目前这个无法遍历 所以我们需要转换一下
    import scala.collection.JavaConverters._
    for ((partition,offset) <- offsets.asScala) {
      //我们将获取到的数据进行封装
      val tp: TopicPartition = new TopicPartition(topic, partition.toInt)
      //然后使用之前的result封装
      result.put(tp,offset.toLong)
    }
    jedis.close()
    result.toMap
  }

}
